/*
 *  Copyright 2013 National Institute of Advanced Industrial Science and Technology
 *  
 *  Licensed under the Apache License, Version 2.0 (the "License");
 *  you may not use this file except in compliance with the License.
 *  You may obtain a copy of the License at
 *  
 *      http://www.apache.org/licenses/LICENSE-2.0
 *  
 *  Unless required by applicable law or agreed to in writing, software
 *  distributed under the License is distributed on an "AS IS" BASIS,
 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *  See the License for the specific language governing permissions and
 *  limitations under the License.
 */

package org.sss.mapreduce;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.io.ObjectStreamClass;
import java.io.Serializable;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Provides access to configuration parameters and others.
 * <p>
 * <code>Configuration</code> object provides access SSS's configuration
 * parameters.  It will be also used to convey SSS client program's parameters
 * to SSS worker nodes.
 */
@SuppressWarnings("serial")
public class Configuration implements Serializable {
  @SuppressWarnings("unused")
  private static Logger logger = LoggerFactory.getLogger(Configuration.class);

  private final HashMap<String, byte[]> hashmap = new HashMap<String, byte[]>();

  // must be null at first, to make this serializable
  private ClassLoader classLoader = null;
  /**
   * Sets a <code>ClassLoader</code> object used with this <code>Configuration</code>.
   *
   * @param  cl  <code>ClassLoader</code> object for this <code>Configuration</code>
   * @see    java.lang.ClassLoader
   */
  public void setClassLoader(ClassLoader cl) {
    this.classLoader = cl;
  }
  /**
   * Gets a <code>ClassLoader</code> object coupled to this <code>Configuration</code>.
   *
   * @return  the <code>ClassLoader</code> object
   */
  public ClassLoader getClassLoader() {
    if (classLoader == null) {
      this.classLoader = this.getClass().getClassLoader(); // default loader
    }
    return this.classLoader;
  }
  /**
   * Indicates whether this <code>Configuration</code> contains a mapping for
   * the specified key.
   *
   * @param  name   name for the key whose presence in this
   * <code>Configuration</code> is to be tested
   * @return  <tt>true</tt> if a variable with the specified name exists;
   * <tt>false</tt> otherwise
   */
  public boolean containsKey(String name) {
    return hashmap.containsKey(name);
  }
  /**
   * Returns the raw value to which the specified key is mapped,
   * or <tt>null</tt> if this <code>Configuration</code> contains no mapping
   * for the key.
   *
   * @param  name  the name of the key whose associated value is to be returned
   * @return the raw value to which the specified key is mapped;
   * or <tt>null</tt> if this <code>Configuration</code> contains no mapping
   * for the key
   */
  public byte[] getRaw(String name) {
    return hashmap.get(name);
  }
  /**
   * Associates the specified raw value with the specified key in this
   * <code>Configuration</code>.
   *
   * @param  name   the name of the key with which the specified value is to be associated
   * @param  value  value to be associated with the specified key
   */
  public void setRaw(String name, byte[] value) {
    hashmap.put(name, value);
  }
  /**
   * Associates the specified <tt>String</tt> value with the specified key
   * in this <code>Configuration</code>.
   *
   * @param  name   the name of the key with which the specified value is to be associated
   * @param  value  value to be associated with the specified key
   * @throws SssException
   */
  public void set(String name, String value) throws SssException {
    setRaw(name, toBytes(value));
  }
  /**
   * Returns the <tt>String</tt> value to which the specified key is mapped,
   * or <tt>null</tt> if this <code>Configuration</code> contains no mapping
   * for the key.
   *
   * @param  name  the name of the key whose associated value is to be returned
   * @return the <tt>String</tt> value to which the specified key is mapped;
   * or <tt>null</tt> if this <code>Configuration</code> contains no mapping
   * for the key
   * @throws SssException
   */
  public String get(String name) throws SssException {
    byte[] bs = getRaw(name);
    if (bs != null) {
      return (String)(fromBytes(bs, null));
    } else {
      return null;
    }
  }
  /**
   * Returns the <tt>String</tt> value to which the specified key is mapped,
   * or <tt>defaultValue</tt> if this <code>Configuration</code> contains no
   * mapping for the key.
   *
   * @param  name          the name of the key whose associated value is to be returned
   * @param  defaultValue  the <tt>String</tt> value used as default value to be returned
   * @return the <tt>String</tt> value to which the specified key is mapped;
   * or <tt>defaultValue</tt> if this <code>Configuration</code> contains no
   * mapping for the key
   * @throws SssException
   */
  public String get(String name, String defaultValue) throws SssException {
    if (containsKey(name)) {
      return get(name);
    } else {
      return defaultValue;
    }
  }
  /**
   * Appends the <tt>String</tt> value to the existing <tt>String</tt> value.
   *
   * @param  name   the name of the key whose associated value is to be appended
   * @param  value  the <tt>String</tt> value to append
   * @throws SssException
   */
  public synchronized void append(String name, String value) throws SssException {
    set(name, get(name) + value);
  }
  /**
   * Associates the specified <tt>float</tt> value with the specified key
   * in this <code>Configuration</code>.
   *
   * @param  name   the name of the key with which the specified value is to be associated
   * @param  value  value to be associated with the specified key
   * @throws SssException
   */
  public void setFloat(String name, float value) throws SssException {
    set(name, Float.toString(value));
  }
  /**
   * Returns the <tt>float</tt> value to which the specified key is mapped.
   *
   * @param  name  the name of the key whose associated value is to be returned
   * @return the <tt>float</tt> value to which the specified key is mapped
   * @throws SssException
   */
  public float getFloat(String name) throws SssException {
    return Float.parseFloat(get(name));
  }
  /**
   * Returns the <tt>float</tt> value to which the specified key is mapped,
   * or <tt>defaultValue</tt> if this <code>Configuration</code> contains no
   * mapping for the key.
   *
   * @param  name          the name of the key whose associated value is to be returned
   * @param  defaultValue  the <tt>float</tt> value used as default value to be returned
   * @return the <tt>float</tt> falue to which the specified key is mapped;
   * or <tt>defaultValue</tt> if this <code>Configuration</code> contains no
   * mapping for the key
   * @throws SssException
   */
  public float getFlaot(String name, float defaultValue) throws SssException {
    if (containsKey(name)) {
      return Float.parseFloat(get(name));
    }
    else {
      return defaultValue;
    }
  }
  /**
   * Add the <tt>float</tt> value to the existing <tt>float</tt> value.
   *
   * @param  name   the name of the key whose associated value is to be added
   * @param  value  the <tt>float</tt> value to add
   * @throws SssException
   */
  public synchronized void addFloat(String name, float value) throws SssException {
    setFloat(name, getFloat(name) + value);
  }
  /**
   * Associates the specified <tt>int</tt> value with the specified key
   * in this <code>Configuration</code>.
   *
   * @param  name   the name of the key with which the specified value is to be associated
   * @param  value  value to be associated with the specified key
   * @throws SssException
   */
  public void setInt(String name, int value) throws SssException {
    set(name, Integer.toString(value));
  }
  /**
   * Returns the <tt>int</tt> value to which the specified key is mapped.
   *
   * @param  name  name of the key whose associated value is to be returned
   * @return the <tt>int</tt> value to which the specified key is mapped
   * @throws SssException
   */
  public int getInt(String name) throws SssException {
    return Integer.parseInt(get(name));
  }
  /**
   * Returns the <tt>int</tt> value to which the specified key is mapped,
   * or <tt>defaultValue</tt> if this <code>Configuration</code> contains no
   * mapping for the key.
   *
   * @param  name          the name of the key whose associated value is to be returned
   * @param  defaultValue  the <tt>int</tt> value used as default value to be returned
   * @return the <tt>int</tt> value to which the specified key is mapped;
   * or <tt>defaultValue</tt> if this <code>Configuration</code> contains no
   * mapping for the key
   * @throws SssException
   */
  public int getInt(String name, int defaultValue) throws SssException {
    if (containsKey(name)) {
      return Integer.parseInt(get(name));
    } else {
      return defaultValue;
    }
  }
  /**
   * Add the <tt>int</tt> value to the existing <tt>int</tt> value.
   *
   * @param  name   the name of the key whose associated value is to be added
   * @param  value  the <tt>int</tt> value to add
   * @throws SssException
   */
  public synchronized void addInt(String name, int value) throws SssException {
    setInt(name, getInt(name) + value);
  }
  /**
   * Associates the specified <tt>long</tt> value with the specified key
   * in this <code>Configuration</code>.
   *
   * @param  name   the name of the key with which the specified value is to be associated
   * @param  value  value to be associated with the specified key
   * @throws SssException
   */
  public void setLong(String name, long value) throws SssException {
    set(name, Long.toString(value));
  }
  /**
   * Returns the <tt>long</tt> value to which the specified key is mapped.
   *
   * @param  name  the name of the key whose associated value is to be returned
   * @return the <tt>long</tt> value to which the specified key is mapped
   * @throws SssException
   */
  public long getLong(String name) throws SssException {
    return Long.parseLong(get(name));
  }
  /**
   * Returns the <tt>long</tt> value to which the specified key is mapped,
   * or <tt>defaultValue</tt> if this <code>Configuration</code> contains no
   * mapping for the key.
   *
   * @param  name          the name of the key whose associated value is to be returned
   * @param  defaultValue  the <tt>long</tt> value used as default value to be returned
   * @return the <tt>long</tt> value to which the specified key is mapped;
   * or <tt>defaultValue</tt> if this <code>Configuration</code> contains no
   * mapping for the key
   * @throws SssException
   */
  public long getLong(String name, long defaultValue) throws SssException {
    if (containsKey(name)) {
      return Long.parseLong(get(name));
    } else {
      return defaultValue;
    }
  }
  /**
   * Add the <tt>long</tt> value to the existing <tt>long</tt> value.
   *
   * @param  name   name of the key whose associated value is to be added
   * @param  value  the <tt>long</tt> value to add
   * @throws SssException
   */
  public synchronized void addLong(String name, int value) throws SssException {
    setLong(name, getLong(name) + value);
  }
  /**
   * Associates the specified <tt>boolean</tt> value with the specified key
   * in this <code>Configuration</code>.
   *
   * @param  name   the name of the key with which the specified value to be associated
   * @param  value  value to be associated with the specified key
   * @throws SssException
   */
  public void setBoolean(String name, boolean value) throws SssException {
    set(name, Boolean.toString(value));
  }
  /**
   * Returns the <tt>boolean</tt> value to which the specified key is mapped.
   *
   * @param  name  the name of the key whose associated value is to be returned
   * @return the <tt>boolean</tt> value to which the specified key is mapped
   * @throws SssException
   */
  public boolean getBoolean(String name) throws SssException {
    return Boolean.parseBoolean(get(name));
  }
  /**
   * Return the <tt>boolean</tt> value to which the specified key is mapped,
   * or <tt>defaultValue</tt> if this <code>Configuration</code> contains no
   * mapping for the key.
   *
   * @param  name          the name of the key whose associated value is to be returned
   * @param  defaultValue  the <tt>boolean</tt> value used as default value to be returned
   * @return the <tt>boolean</tt> value to which the specified key is mapped;
   * or <tt>defaultValue</tt> if this <code>Configuration</code> contains no
   * mapping for the key
   * @throws SssException
   */
  public boolean getBoolean(String name, boolean defaultValue) throws SssException {
    if (containsKey(name)) {
      return getBoolean(name);
    } else {
      return defaultValue;
    }
  }
  /**
   * Associates the specified <tt>Object</tt> value with the specified key
   * in this <code>Configuration</code>.
   *
   * @param  name  the name of the key with which the specified value is to be associated
   * @param  o     value to be associated with the specified key
   * @throws SssException
   */
  public void setObject(String name, Object o) throws SssException {
    setRaw(name, toBytes(o));
  }
  /**
   * Returns the <tt>Object</tt> value to which the specified key is mapped.
   *
   * @param  name  the name of the key whose associated value is to be returned
   * @return the <tt>Object</tt> value to which the specified key is mapped
   * @throws SssException
   */
  public Object getObject(String name) throws SssException {
    byte[] bs = getRaw(name);
    if (bs == null) {
      return null;
    }
    return fromBytes(getRaw(name), classLoader);
  }
  /**
   * Returns a <tt>KeyFormat</tt> object to which one of the specified name
   * is mapped.  The first <tt>KeyFormat</tt> object that matches will be returned,
   * or <tt>KeyFormat.DEFAULT</tt> is this <code>Configuration</code> contains
   * no mapping for the name
   *
   * @param  names  the name or the array of names whose associated <tt>KeyFormat</tt>
   * is to be returned
   * @return a <tt>KeyForamt</tt> object to which one of the specified name is mapped
   * @throws SssException
   * @see    org.sss.mapreduce.KeyFormat
   */
  public KeyFormat getKeyFormat(String... names) throws SssException {
    for (String name: names) {
      if (containsKey(name)) {
        return KeyFormat.valueOf(get(name).toUpperCase());
      }
    }
    return KeyFormat.DEFAULT;
  }
  /**
   * Returns a serialized form of the specified <tt>Object</tt>.
   *
   * @param  o  an <tt>Object</tt> to be serialized
   * @return a serialized form of the specified <tt>Object</tt>
   * @throws SssException
   */
  static public byte[] toBytes(Object o) throws SssException {
    try {
      ByteArrayOutputStream bos = new ByteArrayOutputStream();
      ObjectOutputStream oos = new ObjectOutputStream(bos);
      oos.writeObject(o);
      oos.close();
      return bos.toByteArray();
    } catch (IOException e) {
      throw new SssException(e);
    }
  }
  /**
   * Returns a deserialized form object from the specified array of bytes
   * using the specified <tt>ClassLoader</tt>.
   *
   * @param  bytes   a sequence of bytes to be deserialized into an object
   * @param  loader  <tt>ClassLoader</tt> object which will be used to deserialize an object
   * @throws SssException
   * @see    java.lang.ClassLoader
   */
  static public Object fromBytes(byte [] bytes, ClassLoader loader) throws SssException {
    try {
      ByteArrayInputStream bis = new ByteArrayInputStream(bytes);
      if (loader == null) {
        loader = Configuration.class.getClassLoader();
      }
      ObjectInputStream ois = new ClassLoaderObjectInputStream(loader, bis);
      try {
        return ois.readObject();
      }
      finally {
        ois.close();
      }
    } catch (IOException e) {
      throw new SssException(e);
    } catch (ClassNotFoundException e) {
      throw new SssException(e);
    }
  }
  /**
   * <tt>ObjectInputStream</tt> with controllable <tt>ClassLoader</tt>
   */
  static public class ClassLoaderObjectInputStream extends ObjectInputStream {
    private final ClassLoader classLoader;
    public ClassLoaderObjectInputStream(ClassLoader classLoader, InputStream in) throws IOException {
      super(in);
      this.classLoader = classLoader;
    }

    @Override
    protected Class<?> resolveClass(ObjectStreamClass desc) throws IOException, ClassNotFoundException {
      try{
        String name = desc.getName();
        return Class.forName(name, false, classLoader);
      }
      catch(ClassNotFoundException e) {
        return super.resolveClass(desc);
      }
    }
  }
  /**
   * Copies all of the mappings from the specified <tt>Configuration</tt> object
   * to this <tt>Configuration</tt> object.  These mappings will replace any
   * mappings taht this <tt>Configuration</tt> had for any of the keys currently
   * in the specified <tt>Configuration</tt>.
   *
   * @param  other  the other <tt>Configuration</tt> object
   */
  public void merge(Configuration other) {
    hashmap.putAll(other.hashmap);
  }
  /**
   * Load mappings from a file.
   * <p>
   * The file should be in the form of properties file.
   *
   * @param  filename  path for the properties file to be load
   * @throws SssException
   */
  public void load(String filename) throws SssException {
    try {
      Properties p = new Properties();

      p.load(new FileInputStream(filename));

      for (Map.Entry<Object, Object> e: p.entrySet()) {
        this.set((String)e.getKey(), (String)e.getValue());
      }
    } catch (IOException e) {
      throw new SssException(e);
    }
  }
}
